package com.wjw.redis.stream.handler;

import org.redisson.Redisson;
import org.redisson.api.*;
import org.redisson.config.Config;

import java.util.concurrent.TimeUnit;

/**
 * Redisson 消息队列示例
 * 演示了如何使用 Redisson 实现各种类型的分布式队列
 * ## 示例包含的队列类型
 * 1. 基本队列 (RQueue)
 *
 *    - 最基本的队列实现，提供标准的队列操作
 *    - 演示了 add、offer、poll、peek 等基本操作
 * 2. 阻塞队列 (RBlockingQueue)
 *
 *    - 提供阻塞操作，当队列为空时，获取元素的线程会被阻塞
 *    - 演示了 poll(timeout)、take() 等阻塞方法
 * 3. 有界阻塞队列 (RBoundedBlockingQueue)
 *
 *    - 有容量限制的阻塞队列
 *    - 当队列满时，添加元素的线程会被阻塞
 * 4. 优先级队列 (RPriorityQueue)
 *
 *    - 根据元素的优先级排序，而不是添加顺序
 *    - 通过实现 Comparable 接口来定义优先级
 * 5. 延迟队列 (RDelayedQueue)
 *
 *    - 允许将元素放入队列，并在指定的延迟后才能被消费
 *    - 元素到期后会自动转移到目标队列
 * 6. 生产者-消费者模式
 *
 *    - 使用 Redisson 的队列实现生产者-消费者模式
 *    - 演示了多线程环境下的队列操作
 */
public class RedissonQueueDemo {

    public static void main(String[] args) throws InterruptedException {
        // 创建 Redisson 配置
        Config config = new Config();
        // 使用单节点模式连接 Redis
        config.useSingleServer()
                .setAddress("redis://127.0.0.1:6379")
                .setPassword("redis1105"); // 设置密码，如果没有密码可以去掉这行

        // 创建 Redisson 客户端
        RedissonClient redisson = Redisson.create(config);

        try {
            // 演示不同类型的队列
            demoBasicQueue(redisson);
            demoBlockingQueue(redisson);
            demoBoundedBlockingQueue(redisson);
            demoPriorityQueue(redisson);
            demoDelayedQueue(redisson);
            demoProducerConsumerPattern(redisson);
        } finally {
            // 关闭 Redisson 客户端
            redisson.shutdown();
            System.out.println("Redisson 客户端已关闭");
        }
    }

    /**
     * 演示基本队列 (RQueue)
     * RQueue 是最基本的队列实现，提供了标准的队列操作
     */
    private static void demoBasicQueue(RedissonClient redisson) {
        System.out.println("\n===== 基本队列示例 =====");
        
        // 获取队列实例
        RQueue<String> queue = redisson.getQueue("myBasicQueue");
        
        // 清空队列（用于演示）
        queue.clear();
        
        // 添加元素到队列
        queue.add("消息1");
        queue.add("消息2");
        queue.offer("消息3");
        
        System.out.println("队列大小: " + queue.size());
        System.out.println("队列内容: " + queue);
        
        // 获取并移除队列头部元素
        String firstElement = queue.poll();
        System.out.println("移除的第一个元素: " + firstElement);
        System.out.println("移除后队列大小: " + queue.size());
        
        // 获取但不移除队列头部元素
        String peekElement = queue.peek();
        System.out.println("查看队列头部元素: " + peekElement);
        System.out.println("查看后队列大小: " + queue.size());
    }

    /**
     * 演示阻塞队列 (RBlockingQueue)
     * RBlockingQueue 提供了阻塞操作，当队列为空时，获取元素的线程会被阻塞
     */
    private static void demoBlockingQueue(RedissonClient redisson) throws InterruptedException {
        System.out.println("\n===== 阻塞队列示例 =====");
        
        // 获取阻塞队列实例
        RBlockingQueue<String> blockingQueue = redisson.getBlockingQueue("myBlockingQueue");
        
        // 清空队列（用于演示）
        blockingQueue.clear();
        
        // 启动一个消费者线程
        Thread consumerThread = new Thread(() -> {
            try {
                System.out.println("消费者等待消息...");
                // 阻塞直到有元素可用，或者超时
                String element = blockingQueue.poll(5, TimeUnit.SECONDS);
                System.out.println("消费者接收到消息: " + element);
                
                // 再次阻塞获取
                element = blockingQueue.take();
                System.out.println("消费者接收到消息: " + element);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.out.println("消费者线程被中断");
            }
        });
        consumerThread.start();
        
        // 主线程等待一会儿再生产消息
        Thread.sleep(2000);
        System.out.println("生产者发送消息...");
        blockingQueue.add("阻塞消息1");
        
        Thread.sleep(2000);
        System.out.println("生产者发送另一条消息...");
        blockingQueue.add("阻塞消息2");
        
        // 等待消费者线程完成
        consumerThread.join();
    }

    /**
     * 演示有界阻塞队列 (RBoundedBlockingQueue)
     * RBoundedBlockingQueue 是有容量限制的阻塞队列
     */
    private static void demoBoundedBlockingQueue(RedissonClient redisson) throws InterruptedException {
        System.out.println("\n===== 有界阻塞队列示例 =====");
        
        // 获取有界阻塞队列实例，设置容量为3
        RBoundedBlockingQueue<String> boundedQueue = redisson.getBoundedBlockingQueue("myBoundedQueue");
        boundedQueue.trySetCapacity(3);
        
        // 清空队列（用于演示）
        boundedQueue.clear();
        
        System.out.println("有界队列容量: " + boundedQueue.remainingCapacity());
        
        // 添加元素到队列
        boundedQueue.add("有界消息1");
        boundedQueue.add("有界消息2");
        boundedQueue.add("有界消息3");
        
        System.out.println("队列已满，剩余容量: " + boundedQueue.remainingCapacity());
        
        // 启动一个生产者线程，尝试添加第4个元素（会被阻塞）
        Thread producerThread = new Thread(() -> {
            try {
                System.out.println("生产者尝试添加第4个元素（会被阻塞）...");
                // 这里会阻塞，直到队列有空间
                boundedQueue.put("有界消息4");
                System.out.println("第4个元素已添加到队列");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.out.println("生产者线程被中断");
            }
        });
        producerThread.start();
        
        // 主线程等待一会儿再消费消息
        Thread.sleep(2000);
        System.out.println("消费者移除一个元素...");
        String element = boundedQueue.take();
        System.out.println("移除的元素: " + element);
        
        // 等待生产者线程完成
        producerThread.join();
        
        System.out.println("当前队列大小: " + boundedQueue.size());
    }

    /**
     * 演示优先级队列 (RPriorityQueue)
     * RPriorityQueue 根据元素的优先级排序，而不是添加顺序
     */
    private static void demoPriorityQueue(RedissonClient redisson) {
        System.out.println("\n===== 优先级队列示例 =====");
        
        // 获取优先级队列实例
        RPriorityQueue<PriorityTask> priorityQueue = redisson.getPriorityQueue("myPriorityQueue");
        
        // 清空队列（用于演示）
        priorityQueue.clear();
        
        // 添加任务到优先级队列，优先级不同
        priorityQueue.add(new PriorityTask("低优先级任务", 3));
        priorityQueue.add(new PriorityTask("中优先级任务", 2));
        priorityQueue.add(new PriorityTask("高优先级任务", 1));
        
        System.out.println("优先级队列大小: " + priorityQueue.size());
        
        // 按优先级顺序获取任务
        System.out.println("按优先级顺序获取任务:");
        while (!priorityQueue.isEmpty()) {
            PriorityTask task = priorityQueue.poll();
            System.out.println("- " + task);
        }
    }

    /**
     * 演示延迟队列 (RDelayedQueue)
     * RDelayedQueue 允许将元素放入队列，并在指定的延迟后才能被消费
     */
    private static void demoDelayedQueue(RedissonClient redisson) throws InterruptedException {
        System.out.println("\n===== 延迟队列示例 =====");
        
        // 获取目标队列，延迟队列会将元素转移到这个队列
        RQueue<String> destinationQueue = redisson.getQueue("myDestinationQueue");
        
        // 获取延迟队列实例
        RDelayedQueue<String> delayedQueue = redisson.getDelayedQueue(destinationQueue);
        
        // 清空队列（用于演示）
        destinationQueue.clear();
        
        System.out.println("添加延迟消息...");
        // 添加延迟消息，2秒后可用
        delayedQueue.offer("延迟消息1", 2, TimeUnit.SECONDS);
        // 添加延迟消息，5秒后可用
        delayedQueue.offer("延迟消息2", 5, TimeUnit.SECONDS);
        
        System.out.println("等待延迟消息转移到目标队列...");
        
        // 检查目标队列
        for (int i = 0; i < 6; i++) {
            System.out.println("目标队列大小: " + destinationQueue.size() + 
                             ", 内容: " + destinationQueue);
            Thread.sleep(1000);
        }
        
        // 不再需要时销毁延迟队列
        delayedQueue.destroy();
    }

    /**
     * 演示生产者-消费者模式
     * 使用 Redisson 的队列实现生产者-消费者模式
     */
    private static void demoProducerConsumerPattern(RedissonClient redisson) throws InterruptedException {
        System.out.println("\n===== 生产者-消费者模式示例 =====");
        
        // 获取阻塞队列实例
        RBlockingQueue<String> taskQueue = redisson.getBlockingQueue("myTaskQueue");
        
        // 清空队列（用于演示）
        taskQueue.clear();
        
        // 创建并启动消费者线程
        Thread consumerThread = new Thread(() -> {
            try {
                System.out.println("消费者启动，等待任务...");
                while (!Thread.currentThread().isInterrupted()) {
                    // 阻塞等待任务，超时时间为10秒
                    String task = taskQueue.poll(10, TimeUnit.SECONDS);
                    if (task != null) {
                        System.out.println("消费者处理任务: " + task);
                        // 模拟任务处理时间
                        Thread.sleep(1000);
                    } else {
                        // 超时，没有任务可处理
                        System.out.println("消费者等待超时，退出");
                        break;
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.out.println("消费者线程被中断");
            }
        });
        consumerThread.start();
        
        // 生产者发送任务
        System.out.println("生产者开始发送任务...");
        for (int i = 1; i <= 5; i++) {
            String task = "任务" + i;
            taskQueue.add(task);
            System.out.println("生产者发送任务: " + task);
            Thread.sleep(500);
        }
        
        // 等待消费者处理完所有任务
        consumerThread.join();
    }

    /**
     * 用于优先级队列的任务类
     * 实现 Comparable 接口以支持优先级排序
     */
    private static class PriorityTask implements Comparable<PriorityTask> {
        private final String name;
        private final int priority; // 优先级，数字越小优先级越高
        
        public PriorityTask(String name, int priority) {
            this.name = name;
            this.priority = priority;
        }
        
        @Override
        public int compareTo(PriorityTask other) {
            // 按优先级排序，优先级数字小的排在前面
            return Integer.compare(this.priority, other.priority);
        }
        
        @Override
        public String toString() {
            return name + " (优先级: " + priority + ")";
        }
    }
}